package com.wzy.event;

import com.alibaba.fastjson.JSONObject;
import com.wzy.domain.entity.Event;
import com.wzy.domain.entity.SysMessage;
import com.wzy.mapper.SysMessageMapper;
import com.wzy.util.RedisCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import static com.wzy.constant.RedisConstants.*;
import static com.wzy.constant.SystemConstants.*;

/**
 * 消息消费
 *
 * @author ultima
 * @version 2022/10/26 19:48
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class EventConsumer
{
    private final SysMessageMapper sysMessageMapper;
    private final RedisCache redisCache;

    @KafkaListener(topics = {TOPIC_COMMENT, TOPIC_FOLLOW, TOPIC_LIKE})
    public void handleSysMessage(ConsumerRecord<String, String> consumerRecord)
    {
        if (consumerRecord == null || consumerRecord.value() == null)
        {
            log.error("消息为空！");
            return;
        }
        Event event = JSONObject.parseObject(consumerRecord.value(), Event.class);
        if (event == null)
        {
            log.error("消息格式错误！");
            return;
        }

        // 封装消息
        SysMessage message = new SysMessage();
        message.setFromId(SYSTEM_ID);
        message.setToId(event.getEntityUserId());
        message.setFromUsername(event.getUserNick());
        message.setAvatarUrl(event.getAvatarUrl());
        message.setConversationType(event.getTopic());
        message.setCreateTime(new Date());
        // 消息内容JSON
        Map<String, Object> content = new HashMap<>();
        content.put("userId", event.getUserId());
        content.put("entityType", event.getEntityType());
        content.put("entityId", event.getEntityId());
        content.put("username", event.getUserNick());
        if (!event.getData().isEmpty())
        {
            event.getData().forEach(content::put);
        }
        message.setContent(JSONObject.toJSONString(content));
        sysMessageMapper.insert(message);
        String msgKey;
        switch (event.getTopic())
        {
            case TOPIC_COMMENT:
                msgKey = SYSTEM_MESSAGE_COMMENT_NUM_KEY;
                break;
            case TOPIC_LIKE:
                msgKey = SYSTEM_MESSAGE_LIKE_NUM_KEY;
                break;
            case TOPIC_FOLLOW:
                msgKey = SYSTEM_MESSAGE_FOLLOW_NUM_KEY;
                break;
            default:
                log.error("未知主题: {}", event.getTopic());
                throw new IllegalStateException("未知主题: " + event.getTopic());
        }
        // 未读消息加一
        redisCache.increment(msgKey + event.getEntityUserId());
    }

}
